package com.shujia.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.io.Source

object Demo2StudentProducer {

  def main(args: Array[String]): Unit = {

    val properties = new Properties()

    properties.setProperty("bootstrap.servers","master:9092,node1:9092,node2:9092")

    properties.setProperty("key.serializer","org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer","org.apache.kafka.common.serialization.StringSerializer")

    val producer = new KafkaProducer[String,String](properties)

    val students: List[String] = Source.fromFile("data/students.txt").getLines().toList

    for (student <- students) {
      /**
        * 同一个班级保存在同一个分区中
        */
      val clazz: String = student.split(",")(4)
      val partition: Int = Math.abs(clazz.hashCode % 3)

      val record = new ProducerRecord[String,String]("student",partition,null,student)

      producer.send(record)

      producer.flush()

    }

    producer.close()

  }

}
